package cn.z2huo.demo.kafka.spring.consumer.config;

import cn.z2huo.demo.kafka.spring.consumer.properties.TestTopicProperties;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;

import java.time.Duration;

/**
 * <p>
 *
 * @author z2huo
 */
@RequiredArgsConstructor
@Configuration
@EnableConfigurationProperties(TestTopicProperties.class)
public class TestTopicConsumerConfig {

    private final TestTopicProperties testTopicProperties;

    @Bean(name = "testTopicKafkaConsumerFactory")
    public ConsumerFactory<String, String> kafkaConsumerFactory() {
        DefaultKafkaConsumerFactory<String, String> factory = new DefaultKafkaConsumerFactory<>(testTopicProperties.buildConsumerProperties());
        // 用来指定
        factory.setBeanName("test-topic-kafka-consumer");
        return factory;
    }

    @Bean(name = "testTopicKafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            @Qualifier("testTopicKafkaConsumerFactory") ConsumerFactory<String, String> kafkaConsumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory<>();
        listenerContainerFactory.setConsumerFactory(kafkaConsumerFactory);

        configureListenerFactory(listenerContainerFactory);
        configureContainer(listenerContainerFactory.getContainerProperties());

        listenerContainerFactory.setChangeConsumerThreadName(false);
        listenerContainerFactory.setThreadNameSupplier(container ->
                "test-topic-consumer-" + container.getListenerId());

        listenerContainerFactory.setContainerCustomizer(listenerContainer -> {
            listenerContainer.setAlwaysClientIdSuffix(true);
        });

        return listenerContainerFactory;
    }

    private void configureListenerFactory(ConcurrentKafkaListenerContainerFactory<String, String> factory) {
        PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
        KafkaProperties.Listener properties = this.testTopicProperties.getListener();
        map.from(properties::getConcurrency).to(factory::setConcurrency);
        map.from(properties::isAutoStartup).to(factory::setAutoStartup);
//        map.from(this.batchMessageConverter).to(factory::setBatchMessageConverter);
//        map.from(this.recordMessageConverter).to(factory::setRecordMessageConverter);
//        map.from(this.recordFilterStrategy).to(factory::setRecordFilterStrategy);
//        map.from(this.replyTemplate).to(factory::setReplyTemplate);
        if (properties.getType().equals(KafkaProperties.Listener.Type.BATCH)) {
            factory.setBatchListener(true);
        }
//        map.from(this.commonErrorHandler).to(factory::setCommonErrorHandler);
//        map.from(this.afterRollbackProcessor).to(factory::setAfterRollbackProcessor);
//        map.from(this.recordInterceptor).to(factory::setRecordInterceptor);
//        map.from(this.batchInterceptor).to(factory::setBatchInterceptor);
//        map.from(this.threadNameSupplier).to(factory::setThreadNameSupplier);
        map.from(properties::getChangeConsumerThreadName).to(factory::setChangeConsumerThreadName);
    }

    private void configureContainer(ContainerProperties container) {
        PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
        KafkaProperties.Listener properties = this.testTopicProperties.getListener();
        map.from(properties::getAckMode).to(container::setAckMode);
        map.from(properties::getAsyncAcks).to(container::setAsyncAcks);
        map.from(properties::getClientId).to(container::setClientId);
        map.from(properties::getAckCount).to(container::setAckCount);
        map.from(properties::getAckTime).as(Duration::toMillis).to(container::setAckTime);
        map.from(properties::getPollTimeout).as(Duration::toMillis).to(container::setPollTimeout);
        map.from(properties::getNoPollThreshold).to(container::setNoPollThreshold);
        map.from(properties.getIdleBetweenPolls()).as(Duration::toMillis).to(container::setIdleBetweenPolls);
        map.from(properties::getIdleEventInterval).as(Duration::toMillis).to(container::setIdleEventInterval);
        map.from(properties::getIdlePartitionEventInterval)
                .as(Duration::toMillis)
                .to(container::setIdlePartitionEventInterval);
        map.from(properties::getMonitorInterval)
                .as(Duration::getSeconds)
                .as(Number::intValue)
                .to(container::setMonitorInterval);
        map.from(properties::getLogContainerConfig).to(container::setLogContainerConfig);
        map.from(properties::isMissingTopicsFatal).to(container::setMissingTopicsFatal);
        map.from(properties::isImmediateStop).to(container::setStopImmediate);
        map.from(properties::isObservationEnabled).to(container::setObservationEnabled);

//        map.from(this.transactionManager).to(container::setKafkaAwareTransactionManager);
//        map.from(this.rebalanceListener).to(container::setConsumerRebalanceListener);
//        map.from(this.listenerTaskExecutor).to(container::setListenerTaskExecutor);
    }

}
